package org.tong.kafka.producers;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.Date;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * @author
 * @date 2022-08-01 11:42
 */
//@RestController
public class MyProducer {

    //@Autowired
    KafkaTemplate<String, String> kafkaTemplate;

    @RequestMapping("/kafka")
    public String view() {
        return "welcome to use kafka:" + new Date();
    }

    @RequestMapping("/kafka/send/{topic}/{msg}")
    public String msg(@PathVariable("topic") String topic, @PathVariable("msg") String msg) {
        ListenableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, msg);
        CompletableFuture<SendResult<String, String>> completable = future.completable();
        try {
            completable.get();
            return "发送完成 " + new Date();
        } catch (Exception e) {
            e.printStackTrace();
            return "发送失败 " + new Date();
        }
    }
}
